package job.wuba;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import tools.CheckFileTools;
import tools.MyDateUtil;

public class MiddleNetflowJob_M {
	public static class M1 extends Mapper<LongWritable, Text, Text, NullWritable> {
		public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			Configuration conf=context.getConfiguration();
			String plat=conf.get("plat");
			String runDate=conf.get("runDate");//yyyy-MM-dd
			String[] arrys = value.toString().split("\t");
			String[] splitArry = null;
			String newKey = "";
			String newVal = arrys[18]+"\t"+arrys[19]+"\t"+arrys[20]+"\t"+arrys[21]+"\t"+arrys[22]+"\t"+arrys[23]+"\t"+arrys[24]+"\t"+arrys[25]+"\t"+arrys[26]+"\t"+arrys[27];
			newVal+=arrys[28]+"\t"+arrys[29]+"\t"+arrys[30]+"\t"+arrys[31]+"\t"+arrys[32]+"\t"+arrys[33]+"\t"+arrys[34]+"\t"+arrys[35]+"\t"+arrys[36]+"\t"+arrys[37];
			if(plat.equals("m_netflow_m_pv")){
				splitArry = "6,10,11,12,15,16,17".split(",");
				newKey = runDate+"\t"+arrys[1]+"\t"+arrys[2]+"\t"+arrys[3]+"\t"+arrys[4]+"\t"+arrys[5]+"\t"+arrys[9]+"\t"+arrys[13]+"\t"+arrys[14];
			}
			if(plat.equals("m_netflow_m_city_cate_pv")){
				splitArry = "1,3,4,5,6,12,16,17".split(",");
				newKey = runDate+"\t"+arrys[2]+"\t"+arrys[9]+"\t"+arrys[10]+"\t"+arrys[11]+"\t"+arrys[13]+"\t"+arrys[14]+"\t"+arrys[15];
			}
			if(plat.equals("m_netflow_m_cate_pv")){
				splitArry = "9,10,11,12".split(",");
				newKey = runDate+"\t"+arrys[1]+"\t"+arrys[2]+"\t"+arrys[3]+"\t"+arrys[4]+"\t"+arrys[5]+"\t"+arrys[6]+"\t"+arrys[8]+"\t"+arrys[13]+"\t"+arrys[14]+"\t"+arrys[15]+"\t"+arrys[16]+"\t"+arrys[17];
			}
			if(plat.equals("m_netflow_m_source_pv")){
				splitArry = "9,10,11,12,13,14,15,16,17".split(",");
				newKey = runDate+"\t"+arrys[1]+"\t"+arrys[2]+"\t"+arrys[3]+"\t"+arrys[4]+"\t"+arrys[5]+"\t"+arrys[6];
			}
			boolean flag = true;
			for (String split : splitArry) {
				if(!arrys[Integer.parseInt(split)].equals("all")){
					flag = false;
					break;
				}
			}
			if(flag){
				 context.write(new Text(newKey+"\t"+newVal), NullWritable.get());
			}
		}
	}
	public static class R1 extends Reducer<Text,Text,Text,NullWritable>
	{
		public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException
		{
			context.write(key, NullWritable.get());

		}
	}
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, ParseException
	{
		// 获取main输入参数
		Configuration conf = new Configuration();
		//禁止压缩文件输出
		//conf.setBoolean("mapreduce.output.fileoutputformat.compress", false);
		String[] otherArgs =new GenericOptionsParser(conf, args).getRemainingArgs();

		Date currentTime = new Date();
		SimpleDateFormat smdf = new SimpleDateFormat("yyyyMMdd");
		String yestoday = smdf.format(currentTime.getTime() - 1000 * 60 * 60 * 24);
		String startDate = "";
		String endDate = "";
		String[] platArr = null;
		if(otherArgs.length>2){
			startDate = otherArgs[0];
			endDate = otherArgs[1];
			platArr = otherArgs[2].split(",");
		}else{
			startDate = yestoday;
			endDate = yestoday;
			platArr = "m_netflow_m_pv,m_netflow_m_city_cate_pv,m_netflow_m_cate_pv,m_netflow_m_source_pv".split(",");
		}
		int returnCode = 126;
		for (String plat : platArr) {
			for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
				Date date = new SimpleDateFormat("yyyyMMdd").parse(runDate);
				String now = new SimpleDateFormat("yyyy-MM-dd").format(date);
				conf.set("runDate", now);
				conf.set("plat", plat);
				Job job = new Job(conf, "MiddleNetflowJob_M");
				job.setJarByClass(MiddleNetflowJob_M.class);
				job.setMapperClass(M1.class);
				job.setReducerClass(R1.class);
				job.setNumReduceTasks(100);
				job.setOutputKeyClass(Text.class);
				job.setOutputValueClass(NullWritable.class);
				job.setMapOutputKeyClass(Text.class);
				job.setMapOutputValueClass(NullWritable.class);
				//job.setInputFormatClass(TextInputFormat.class);
				//-------------------------DP---------------------------------------
				String outPath = "/home/hdp_lbg_ecdata_dw/resultdata/caods/middle/"+plat+"/"+runDate;
				List<String> inputList = new ArrayList<String>();
				inputList.add("/home/hdp_lbg_ecdata_dw/resultdata/caods/netflow/track.58.com/m/" + runDate) ;
				FileSystem.get(conf).delete(new Path(outPath), true);
				FileOutputFormat.setOutputPath(job, new Path(outPath));


				for(String ifPath:inputList)
				{
					System.out.println("###########输入路径#############"+ifPath);
					CheckFileTools.myJobWait(ifPath, 100, 300000, 30);
					FileInputFormat.addInputPath(job, new Path(ifPath));
				}
				returnCode = job.waitForCompletion(true) ? 0 : 1;
				if (returnCode != 0)
				{
					System.exit(returnCode);
				}
				System.out.println("**********************	end	"+plat+"---"+runDate+"+*********************");
			}
		}
		System.exit(returnCode);
	}
}
